package com.leilei.join.window;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.java.tuple.Tuple2;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/27 21:52
 * @desc 自定义车辆类型水位线设置 默认以系统时间为时间时间
 */
public class VehicleTypeWatermark implements WatermarkStrategy<Tuple2<Integer, String>> {
    @Override
    public TimestampAssigner<Tuple2<Integer, String>> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
        return (element, recordTimestamp) -> System.currentTimeMillis();
    }

    @Override
    public WatermarkGenerator<Tuple2<Integer, String>> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
        return new WatermarkGenerator<Tuple2<Integer, String>>() {
            @Override
            public void onEvent(Tuple2<Integer, String> event, long eventTimestamp, WatermarkOutput output) {
                output.emitWatermark(new Watermark(System.currentTimeMillis()));
            }

            @Override
            public void onPeriodicEmit(WatermarkOutput output) {
                output.emitWatermark(new Watermark(System.currentTimeMillis()));
            }
        };
    }
}